package com.ctgu.rabbitmq.rabbitmq.idempotent;

import com.rabbitmq.client.Channel;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.time.Duration;

/**
 * @ ClassName IdempotentConsumer
 * @ Author Alex2
 * @ Date 2025/8/18 14:53
 **/
@Log4j2
@Component
public class IdempotentConsumer {

    private final StringRedisTemplate redisTemplate;


    public IdempotentConsumer(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @RabbitListener(queues = IdempotentConfig.IDEMPOTENT_QUEUE)
    public void onMessage(Message message, Channel channel) throws Exception {
        Object msgIdObj = message.getMessageProperties().getHeader("messageId");
        if (msgIdObj == null || "".equals(msgIdObj.toString())) {
            log.error("消息没有唯一ID，无法保证幂等性，body={}", new String(message.getBody(), StandardCharsets.UTF_8));
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            return;
        }
        String msgId = msgIdObj.toString();

        try {
            // SETNX + TTL 保证幂等性
            Boolean success = redisTemplate.opsForValue().setIfAbsent(msgId, "1", Duration.ofHours(1));

            if (Boolean.TRUE.equals(success)) {
                // 消息未处理过，执行业务逻辑
                String body = new String(message.getBody(), StandardCharsets.UTF_8);
                log.info("处理消息: {} , msgId={}", body, msgId);
                // 手动确认消息
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } else {
                // 重复消息，直接 ack 忽略
                log.warn("重复消息，已忽略: msgId={}", msgId);
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }
        } catch (Exception e) {
            log.error("处理消息异常, msgId={}, body={}", msgId, new String(message.getBody(), StandardCharsets.UTF_8), e);
            // 不 ack，让 RabbitMQ 重试
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
}
